package cn.itcast.dmp.runner

import cn.itcast.dmp.DmpApp.logInfo
import cn.itcast.dmp.config.AppConfigHelper
import cn.itcast.dmp.etl.IPProcessor
import org.apache.spark.SparkConf
import org.apache.spark.sql.{Column, DataFrame, SparkSession}

object PmtETLRunner {
  /*** 1. 创建 SparkSession * 2. 读取数据集 * 3. 数据操作 * 4. 数据落地 */
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = {
      // a. 创建SparkConf实例对象，设置应用名称
      var sparkConf: SparkConf = new SparkConf().setAppName(this.getClass.getSimpleName.stripSuffix("$"))
      // b. 判断应用运行模式（本地模式还是集群模式，如是本地模式，直接设置值；如是集群模式，通过spark-submit --master）
      if (AppConfigHelper.SPARK_APP_LOCAL_MODE.toBoolean) {
        //如果是本地模式，就选取application.conf文件中的spark.master=local[4]
        sparkConf.setMaster(AppConfigHelper.SPARK_APP_MASTER)
      }
      // c. 建造者模式创建SparkSession实例对象
      import cn.itcast.dmp.config.SparkConfigHelper._
      val session: SparkSession = SparkSession.builder()
        .config(sparkConf) //因为config返回的是一个build对象，所以要加载其他配置信息必须是build类型，
        // 加载应用配置信息， 通过隐式转换（自定义隐式转换函数实现） 从而实现动态加载属性
        .loadSparkConf()
        .getOrCreate()

      // d. 返回会话实例
      session
    }
    import spark.implicits._
    // 2. 读取数据集
    val pmtDF: DataFrame = spark.read.json(AppConfigHelper.AD_DATA_PATH)
//    pmtDF.printSchema()
//    pmtDF.show(10,truncate = false)
    // 开发测试，为了查看WEB UI监控4040端口，让线程休眠
    Thread.sleep(100000)

    // 3. 数据操作（ETL操作）
    val etlDF: DataFrame =IPProcessor.processData(pmtDF)
    etlDF.show(10, truncate = false)
    // 对DataFrame选取需要的字段
    val selectColumns: Seq[Column] = Seq(
      'sessionid, 'advertisersid, 'adorderid, 'adcreativeid,
      'adplatformproviderid, 'sdkversion, 'adplatformkey,
      'putinmodeltype, 'requestmode, 'adprice, 'adppprice,
      'requestdate, 'ip, 'appid, 'appname, 'uuid, 'device,
      'client, 'osversion, 'density, 'pw, 'ph, 'longitude,
      'latitude, 'province, 'city, 'ispid, 'ispname, 'networkmannerid,
      'networkmannername, 'iseffective, 'isbilling, 'adspacetype,
      'adspacetypename, 'devicetype, 'processnode, 'apptype,
      'district, 'paymode, 'isbid, 'bidprice, 'winprice,
      'iswin, 'cur, 'rate, 'cnywinprice, 'imei, 'mac, 'idfa,
      'openudid, 'androidid, 'rtbprovince, 'rtbcity, 'rtbdistrict,
      'rtbstreet, 'storeurl, 'realip, 'isqualityapp, 'bidfloor, 'aw,
      'ah, 'imeimd5, 'macmd5, 'idfamd5, 'openudidmd5, 'androididmd5,
      'imeisha1, 'macsha1, 'idfasha1, 'openudidsha1, 'androididsha1,
      'uuidunknow, 'userid, 'reqdate, 'reqhour, 'iptype, 'initbidprice,
      'adpayment, 'agentrate, 'lomarkrate, 'adxrate, 'title, 'keywords,
      'tagid, 'callbackdate, 'channelid, 'mediatype, 'email, 'tel,
      'age, 'sex,'geoHash
    )
    // 选取字段信息
    val odsDF: DataFrame = etlDF.drop("provincename","cityname", "lang", "lat")
//    / 4. 数据落地 - 存储Kudu表
    import cn.itcast.dmp.untils.KuduUtils._
    // a. 创建表, 表的名称：ods_adinfo_20190914
spark.createKuduTable(AppConfigHelper.AD_MAIN_TABLE_NAME,odsDF.schema,Seq("uuid"))
    // b. 保存数据到Kudu
    odsDF.saveAsKuduTable(AppConfigHelper.AD_MAIN_TABLE_NAME)
    // 开发测试，为了查看WEB UI监控4040端口，让线程休眠
//    Thread.sleep(100000)

    // 当应用运行完成，关闭资源
    spark.stop()
  }

}
